package com.atguigu.day03;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @description :
 * @autor :lyh
 */
// create database userbehavior
// use userbehacior
// create table clicks(username varchar(100),url varchar(100));
public class Example4 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .addSink(new MyJDBC());

        env.execute();
    }

    private static class MyJDBC extends RichSinkFunction<ClickEvent> {
        private Connection connection;
        private PreparedStatement insertStatement;
        private PreparedStatement updateStatement;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection(
                    "jdbc:mysql://localhost:3306/userbehavior?useSSL=false",
                    "root",
                    "19990628"
            );

            insertStatement = connection.prepareStatement(
                    "insert into clicks (username,url) values (?,?)"
            );

            updateStatement = connection.prepareStatement(
                    "update clicks set url = ? where username = ?"
            );
        }


        // 每来一条数据，触发一次执行
        @Override
        public void invoke(ClickEvent in, Context context) throws Exception {
           updateStatement.setString(1,in.url);
           updateStatement.setString(2,in.username);
           updateStatement.execute();

            if (updateStatement.getUpdateCount() == 0) {
                insertStatement.setString(1,in.username);
                insertStatement.setString(2,in.url);
                insertStatement.execute();
            }

        }

        @Override
        public void close() throws Exception {
            insertStatement.close();
            updateStatement.close();
            connection.close();
        }

    }
}
